package my.storm.zmq;

import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class ZmqBolt implements IRichBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 2014797436005766203L;
	
	private Logger LOG = LoggerFactory.getLogger(this.getClass());
	
//	private OutputCollector collector;
	private ZMQ.Context zmqContext;
	private ZMQ.Socket zmqSocket;

	@SuppressWarnings("rawtypes")
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		
//		this.collector = collector;
		
		this.zmqContext = ZMQ.context(1);
		this.zmqSocket = this.zmqContext.socket(ZMQ.PUSH);
		this.zmqSocket.connect("tcp://211.113.22.153:5558");
		
		LOG.info(">>>>> zmqContext={}, zmqSocket={}", this.zmqContext, this.zmqSocket);
		LOG.info(">>>>> ZmqBolt({}) prepared", Thread.currentThread().getId());
	}

	public void execute(Tuple input) {

		LOG.debug(">>>>> execute input={}", input);
		
		byte[] byteReply = input.getBinary(0);
		LOG.info(">>> byteReply={}", byteReply);
		
		this.zmqSocket.send(byteReply, byteReply.length);
//		this.collector.emit(new Values(byteReceived));
	}

	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {

		declarer.declare(new Fields("byteReply"));
	}
}
